package com.caul.demo.hbase;

import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.HColumnDescriptor;
import org.apache.hadoop.hbase.HTableDescriptor;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.util.Bytes;

import java.io.IOException;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;

/**
 * Created by sdliang on 2018/4/18.
 */
public class HBaseTools {

  /**
   * 获取链接
   *
   * @param conf 配置文件
   * @return
   * @throws IOException
   */
  public static Connection getConn(String conf) throws IOException {
    Configuration configuration = HBaseConfiguration.create();
    if (StringUtils.isNotEmpty(conf)) {
      configuration.addResource(conf);
    }
    return ConnectionFactory.createConnection(configuration);
  }

  /**
   * 关闭链接
   *
   * @param conn
   * @throws IOException
   */
  public static void closeConn(Connection conn) throws IOException {
    if (!conn.isClosed()) {
      conn.close();
    }
  }

  /**
   * 创建表
   *
   * @param conn
   * @param tableName   表名
   * @param colFamilies 列族
   * @throws IOException
   */
  public static void createTable(Connection conn, String tableName, String... colFamilies)
      throws IOException {
    Admin admin = conn.getAdmin();
    try {
      HTableDescriptor tabDesc = new HTableDescriptor(TableName.valueOf(tableName));
      HColumnDescriptor colFamily;
      for (String col : colFamilies) {
        colFamily = new HColumnDescriptor(col);
        //设置备份的版本数,默认是3个
        colFamily.setMinVersions(1);
        colFamily.setMaxVersions(3);
        tabDesc.addFamily(colFamily);
      }
      admin.createTable(tabDesc);
    } finally {
      admin.close();
    }
  }

  /**
   * 删除表
   *
   * @param conn
   * @param tableName
   * @throws IOException
   */
  public static void dropTable(Connection conn, String tableName) throws IOException {
    Admin admin = conn.getAdmin();
    try {
      TableName table = TableName.valueOf(tableName);
      //删除前,要禁用
      admin.disableTable(table);
      admin.deleteTable(table);
    } finally {
      admin.close();
    }
  }

  /**
   * 写入或更新一行数据
   *
   * @param conn
   * @param tableName 表名
   * @param rowData   行
   * @throws IOException
   */
  public static void insertAndUpdate(Connection conn, String tableName, HBaseRow rowData)
      throws IOException {
    Table table = conn.getTable(TableName.valueOf(tableName));
    try {
      Put row = new Put(Bytes.toBytes(rowData.getRowId()));
      for (Map.Entry<HBaseRow.HBaseColumn, String> entry : rowData.getData().entrySet()) {
        row.addColumn(Bytes.toBytes(entry.getKey().getFamily()),
            Bytes.toBytes(entry.getKey().getName()),
            Bytes.toBytes(entry.getValue()));
      }
      table.put(row);
    } finally {
      table.close();
    }
  }

  /**
   * 删除一行数据
   *
   * @param conn
   * @param tableName 表名
   * @param rowId     行键
   * @throws IOException
   */
  public static void deleteRow(Connection conn, String tableName, String rowId) throws IOException {
    Table table = conn.getTable(TableName.valueOf(tableName));
    try {
      table.delete(new Delete(Bytes.toBytes(rowId)));
    } finally {
      table.close();
    }
  }

  /**
   * 获取单行记录
   *
   * @param conn
   * @param tableName 表名
   * @param rowId     行键
   * @return
   * @throws IOException
   */
  public static HBaseRow getById(Connection conn, String tableName, String rowId)
      throws IOException {
    Table table = conn.getTable(TableName.valueOf(tableName));
    try {
      return getRow(rowId, table.get(new Get(Bytes.toBytes(rowId))));
    } finally {
      table.close();
    }
  }

  /**
   * 查询数据
   *
   * @param conn
   * @param tableName 表名
   * @param rowBegin  起始行键
   * @param rowEnd    结束行健
   * @return
   * @throws IOException
   */
  public static List<HBaseRow> queryRow(Connection conn, String tableName, String rowBegin,
                                        String rowEnd) throws IOException {
    return queryRow(conn,
        tableName,
        new Scan(Bytes.toBytes(rowBegin), Bytes.toBytes(rowEnd + "0x00")));
  }

  /**
   * 查询数据
   *
   * @param conn
   * @param tableName 表名
   * @param rowBegin  起始行键
   * @param filter    过滤器(
   *                  {@link org.apache.hadoop.hbase.filter.RowFilter},
   *                  {@link org.apache.hadoop.hbase.filter.FamilyFilter},
   *                  {@link org.apache.hadoop.hbase.filter.QualifierFilter},
   *                  {@link org.apache.hadoop.hbase.filter.ValueFilter}
   *                  )
   * @return
   * @throws IOException
   */
  public static List<HBaseRow> queryRow(Connection conn, String tableName, String rowBegin,
                                        Filter filter) throws IOException {
    return queryRow(conn, tableName, new Scan(Bytes.toBytes(rowBegin), filter));
  }

  /**
   * 查询数据
   *
   * @param conn
   * @param tableName 表名
   * @param scan      条件
   * @return
   * @throws IOException
   */
  private static List<HBaseRow> queryRow(Connection conn, String tableName, Scan scan)
      throws IOException {
    Map<String, HBaseRow> rowMap = new HashMap<>();

    Table table = conn.getTable(TableName.valueOf(tableName));
    ResultScanner scanner = table.getScanner(scan);
    try {
      scanner.forEach(result -> {
        while (result.advance()) {
          //rowid
          String rowId = getRowId(result.current());
          //put into map
          if (!rowMap.containsKey(rowId)) {
            rowMap.put(rowId, new HBaseRow(rowId));
            rowMap.get(rowId).setData(new HashMap<>());
          }
          //columnFamily,columnName
          HBaseRow.HBaseColumn key = new HBaseRow.HBaseColumn(getFamily(result.current()),
              getQualifier(result.current()));
          //column value
          rowMap.get(rowId).getData().put(key, getValue(result.current()));
        }
      });
    } finally {
      scanner.close();
      table.close();
    }

    return rowMap.values().parallelStream().collect(Collectors.toList());
  }

  /**
   * 转换一行数据
   *
   * @param rowId  行键
   * @param result
   * @return
   * @throws IOException
   */
  private static HBaseRow getRow(String rowId, Result result) throws IOException {
    HBaseRow row = new HBaseRow(rowId);
    row.setData(new HashMap<>());

    while (result.advance()) {
      //rowid判断
      //      String rowNow = getRowId(result.current());
      //      if (!rowId.equals(rowNow)) {
      //        continue;
      //      }
      //columnFamily,columnName
      HBaseRow.HBaseColumn key = new HBaseRow.HBaseColumn(getFamily(result.current()),
          getQualifier(result.current()));
      //column value
      row.getData().put(key, getValue(result.current()));
    }
    return row;
  }

  /**
   * 获取行键
   *
   * @param cell
   * @return
   */
  private static String getRowId(Cell cell) {
    return Bytes.toString(cell.getRowArray(), cell.getRowOffset(), cell.getRowLength());
  }

  /**
   * 获取列族名
   *
   * @param cell
   * @return
   */
  private static String getFamily(Cell cell) {
    return Bytes.toString(cell.getFamilyArray(), cell.getFamilyOffset(), cell.getFamilyLength());
  }

  /**
   * 获取列名
   *
   * @param cell
   * @return
   */
  private static String getQualifier(Cell cell) {
    return Bytes
        .toString(cell.getQualifierArray(), cell.getQualifierOffset(), cell.getQualifierLength());
  }

  /**
   * 获取列值
   *
   * @param cell
   * @return
   */
  private static String getValue(Cell cell) {
    return Bytes.toString(cell.getValueArray(), cell.getValueOffset(), cell.getValueLength());
  }

}
